package com.xiaoliangkou.data.rdd

import java.sql.{ResultSet, DriverManager}

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.apache.spark.rdd.JdbcRDD
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.{LoggerFactory, Logger}

/**
  * Author: denghp
  * Date: 16-1-26
  * Time: 下午2:46
  * Description: 
  * <p>
  * </p>
  * Version \$version$
  */
object SparkToJDBC {

  val logger = LoggerFactory.getLogger(SparkToJDBC.getClass)
  val objectMapper = new ObjectMapper()
  objectMapper.registerModule(DefaultScalaModule)

  def main(args: Array[String]) {

    val conf = new SparkConf
    val jdbcUrl = "jdbc:mysql://192.168.1.16:3306/cn_data"
    val sc = new SparkContext("spark://master:7077", "SparkToJDBC", conf);
    val stats_path = "/cn_data/xsdata/stats"
    def createConnection() = {
      Class.forName("com.mysql.jdbc.Driver").newInstance()
      DriverManager.getConnection(jdbcUrl, "root", "123456")
    }

    def extractValues(r: ResultSet) = {
      (r.getString(1), r.getString(2), r.getString("PROVINCE"), r.getString("CITY"), r.getString("AREA"))
    }

    val data = new JdbcRDD(sc, createConnection, "select sum(cnt)cnt,sum(my)my, PROVINCE,CITY,AREA,TERMINAL from cn_data.xsdata d where isbn = '9787112159970' and fixprice = 59 and province = '湖南' and CITY = '长沙市' and source = '线下'  and SALEDATE >= '2014-03-25' and SALEDATE <='2014-03-31' " +
      "GROUP BY PROVINCE,CITY,AREA,TERMINAL ORDER BY PROVINCE,CITY,AREA limit ?, ?", lowerBound = 0, upperBound = 100, numPartitions = 1, mapRow = extractValues)

    logger.info("println toList ................")
    println("list : " + data.collect().toList)

    data.collect().foreach(row => logger.info("cnt : {} , my : {}, province : {}, city : {}", row._1, row._2, row._3, row._4))

    logger.info("objectMapper writeValueAsString ................")
    logger.info("stats : {}", objectMapper.writeValueAsString(data.collect().toList))

    logger.info("application successfully stop ................")


    data.saveAsTextFile(stats_path + "/" + System.currentTimeMillis())
    sc.stop()

  }
}
